package com.lenovo.userprofile

import org.apache.hadoop.hbase.HBaseConfiguration
import org.apache.hadoop.hbase.client.Put
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.mapred.TableOutputFormat
import org.apache.hadoop.hbase.util.Bytes
import org.apache.hadoop.mapred.JobConf
import org.apache.spark.sql.SparkSession


object ETL_SET_NULL {
  def main(args: Array[String]): Unit = {

    val sparkSession = SparkSession.builder.master("yarn").appName("ETL_SET_NULL").enableHiveSupport().getOrCreate()
    val hbase_conf = HBaseConfiguration.create()
    val tablename = "upp:upp_user_profile"
    val jobConf = new JobConf(hbase_conf)
    jobConf.setOutputFormat(classOf[TableOutputFormat])
    jobConf.set(TableOutputFormat.OUTPUT_TABLE, tablename)

    val all_user_df= sparkSession.sql("select lower(user_name) from ccsd.ad_user_upp")

    all_user_df.rdd.map(item=>{

      val put = new Put(Bytes.toBytes(item(0)+""))//save's itcode

      put.addColumn(Bytes.toBytes("nature"), Bytes.toBytes("employee_type"), Bytes.toBytes("NULL"))
      put.addColumn(Bytes.toBytes("nature"), Bytes.toBytes("working_years"), Bytes.toBytes("NULL"))
      put.addColumn(Bytes.toBytes("nature"), Bytes.toBytes("band"), Bytes.toBytes("NULL"))
      put.addColumn(Bytes.toBytes("nature"), Bytes.toBytes("bu"), Bytes.toBytes("NULL"))
      put.addColumn(Bytes.toBytes("nature"), Bytes.toBytes("pcg_dcg"), Bytes.toBytes("NULL"))

      put.addColumn(Bytes.toBytes("nature"), Bytes.toBytes("segment"), Bytes.toBytes("NULL"))
      put.addColumn(Bytes.toBytes("nature"), Bytes.toBytes("is_new_user"), Bytes.toBytes("N"))
      put.addColumn(Bytes.toBytes("nature"), Bytes.toBytes("title_change"), Bytes.toBytes("N"))
      put.addColumn(Bytes.toBytes("nature"), Bytes.toBytes("manager"), Bytes.toBytes("N"))
      put.addColumn(Bytes.toBytes("nature"), Bytes.toBytes("is_rol_user"), Bytes.toBytes("N"))

      put.addColumn(Bytes.toBytes("nature"), Bytes.toBytes("is_monthly_user"), Bytes.toBytes("N"))
      put.addColumn(Bytes.toBytes("behavior"), Bytes.toBytes("app_ticket_num"), Bytes.toBytes("NULL"))
      put.addColumn(Bytes.toBytes("behavior"), Bytes.toBytes("not_app_ticket_num"), Bytes.toBytes("NULL"))
      put.addColumn(Bytes.toBytes("behavior"), Bytes.toBytes("ticket_type"), Bytes.toBytes("NULL"))
      put.addColumn(Bytes.toBytes("behavior"), Bytes.toBytes("ticket_distribution"), Bytes.toBytes("NULL"))

      put.addColumn(Bytes.toBytes("behavior"), Bytes.toBytes("peak_day_user"), Bytes.toBytes("NULL"))
      put.addColumn(Bytes.toBytes("behavior"), Bytes.toBytes("peak_week_user"), Bytes.toBytes("NULL"))
      put.addColumn(Bytes.toBytes("behavior"), Bytes.toBytes("peak_month_user"), Bytes.toBytes("NULL"))
      put.addColumn(Bytes.toBytes("behavior"), Bytes.toBytes("way_touch_it"), Bytes.toBytes("NULL"))
      put.addColumn(Bytes.toBytes("behavior"), Bytes.toBytes("ticket_sr"), Bytes.toBytes("N"))
      //
      put.addColumn(Bytes.toBytes("behavior"), Bytes.toBytes("ticket_qi"), Bytes.toBytes("N"))
      put.addColumn(Bytes.toBytes("attitude"), Bytes.toBytes("urgency"), Bytes.toBytes("NULL"))
      put.addColumn(Bytes.toBytes("attitude"), Bytes.toBytes("complaint"), Bytes.toBytes("NULL"))
      put.addColumn(Bytes.toBytes("attitude"), Bytes.toBytes("praise"), Bytes.toBytes("NULL"))
      put.addColumn(Bytes.toBytes("l0_document_property"), Bytes.toBytes("reading_times"), Bytes.toBytes("NULL"))

      put.addColumn(Bytes.toBytes("nature"), Bytes.toBytes("vip"), Bytes.toBytes("N"))

      put.addColumn(Bytes.toBytes("knowledge_level"), Bytes.toBytes("internal_knowledge_level"), Bytes.toBytes("NULL"))
      put.addColumn(Bytes.toBytes("knowledge_level"), Bytes.toBytes("not_internal_knowledge_level"), Bytes.toBytes("NULL"))


      //new numbers 15
      put.addColumn(Bytes.toBytes("behavior"), Bytes.toBytes("CN_peak_day_user"), Bytes.toBytes("NULL"))
      put.addColumn(Bytes.toBytes("behavior"), Bytes.toBytes("CN_peak_week_user"), Bytes.toBytes("NULL"))
      put.addColumn(Bytes.toBytes("behavior"), Bytes.toBytes("ticket_trend"), Bytes.toBytes("NULL"))
      put.addColumn(Bytes.toBytes("behavior"), Bytes.toBytes("CN_peak_month_user"), Bytes.toBytes("NULL"))
      put.addColumn(Bytes.toBytes("behavior"), Bytes.toBytes("assgind_group"), Bytes.toBytes("NULL"))

      put.addColumn(Bytes.toBytes("l0_document_property"), Bytes.toBytes("doc_type"), Bytes.toBytes("NULL"))

      put.addColumn(Bytes.toBytes("system"), Bytes.toBytes("ECC_system_logo"), Bytes.toBytes("N"))
      put.addColumn(Bytes.toBytes("system"), Bytes.toBytes("CIP_system_logo"), Bytes.toBytes("N"))
      put.addColumn(Bytes.toBytes("system"), Bytes.toBytes("CSP_system_logo"), Bytes.toBytes("N"))
      put.addColumn(Bytes.toBytes("system"), Bytes.toBytes("CGP_system_logo"), Bytes.toBytes("N"))
      put.addColumn(Bytes.toBytes("system"), Bytes.toBytes("SRM_EBP_system_logo"), Bytes.toBytes("N"))
      put.addColumn(Bytes.toBytes("system"), Bytes.toBytes("SRM_SUS_system_logo"), Bytes.toBytes("N"))
      put.addColumn(Bytes.toBytes("system"), Bytes.toBytes("PP_system_logo"), Bytes.toBytes("N"))

      //10/17
      put.addColumn(Bytes.toBytes("system"), Bytes.toBytes("ecc_key_t_code"), Bytes.toBytes("NULL"))
      put.addColumn(Bytes.toBytes("system"), Bytes.toBytes("gts_key_t_code"), Bytes.toBytes("NULL"))

      put.addColumn(Bytes.toBytes("system"), Bytes.toBytes("ECC_freq"), Bytes.toBytes("NULL"))
      put.addColumn(Bytes.toBytes("system"), Bytes.toBytes("SRM_EBP_freq"), Bytes.toBytes("NULL"))
      put.addColumn(Bytes.toBytes("system"), Bytes.toBytes("SRM_SUS_freq"), Bytes.toBytes("NULL"))
      put.addColumn(Bytes.toBytes("system"), Bytes.toBytes("PP_freq"), Bytes.toBytes("NULL"))
      put.addColumn(Bytes.toBytes("system"), Bytes.toBytes("BW_freq"), Bytes.toBytes("NULL"))
      put.addColumn(Bytes.toBytes("system"), Bytes.toBytes("GTS_freq"), Bytes.toBytes("NULL"))

      put.addColumn(Bytes.toBytes("system"), Bytes.toBytes("ecc_day_operation_peak"), Bytes.toBytes("NULL"))
      put.addColumn(Bytes.toBytes("system"), Bytes.toBytes("ecc_week_operation_peak"), Bytes.toBytes("NULL"))
      put.addColumn(Bytes.toBytes("system"), Bytes.toBytes("ecc_month_operation_peak"), Bytes.toBytes("NULL"))

      //
      put.addColumn(Bytes.toBytes("system"), Bytes.toBytes("srm_ebp_day_operation_peak"), Bytes.toBytes("NULL"))
      put.addColumn(Bytes.toBytes("system"), Bytes.toBytes("srm_ebp_week_operation_peak"), Bytes.toBytes("NULL"))
      put.addColumn(Bytes.toBytes("system"), Bytes.toBytes("srm_ebp_month_operation_peak"), Bytes.toBytes("NULL"))

      put.addColumn(Bytes.toBytes("system"), Bytes.toBytes("srm_sus_day_operation_peak"), Bytes.toBytes("NULL"))
      put.addColumn(Bytes.toBytes("system"), Bytes.toBytes("srm_sus_week_operation_peak"), Bytes.toBytes("NULL"))
      put.addColumn(Bytes.toBytes("system"), Bytes.toBytes("srm_sus_month_operation_peak"), Bytes.toBytes("NULL"))

      put.addColumn(Bytes.toBytes("system"), Bytes.toBytes("pp_day_operation_peak"), Bytes.toBytes("NULL"))
      put.addColumn(Bytes.toBytes("system"), Bytes.toBytes("pp_week_operation_peak"), Bytes.toBytes("NULL"))
      put.addColumn(Bytes.toBytes("system"), Bytes.toBytes("pp_month_operation_peak"), Bytes.toBytes("NULL"))

      put.addColumn(Bytes.toBytes("system"), Bytes.toBytes("bw_day_operation_peak"), Bytes.toBytes("NULL"))
      put.addColumn(Bytes.toBytes("system"), Bytes.toBytes("bw_week_operation_peak"), Bytes.toBytes("NULL"))
      put.addColumn(Bytes.toBytes("system"), Bytes.toBytes("bw_month_operation_peak"), Bytes.toBytes("NULL"))

      put.addColumn(Bytes.toBytes("system"), Bytes.toBytes("gts_day_operation_peak"), Bytes.toBytes("NULL"))
      put.addColumn(Bytes.toBytes("system"), Bytes.toBytes("gts_week_operation_peak"), Bytes.toBytes("NULL"))
      put.addColumn(Bytes.toBytes("system"), Bytes.toBytes("gts_month_operation_peak"), Bytes.toBytes("NULL"))

      (new ImmutableBytesWritable, put)
    }).saveAsHadoopDataset(jobConf)
 }
}
